"""
OPC UA协议接口实现
"""


import asyncio
import threading
import time
from typing import Dict, Any, Optional, List, Union, Tuple, Callable
from collections import defaultdict
from functools import wraps

from opcua.common.node import Node
from ....core.communication_interface import CommunicationInterface, CommunicationProtocol, ConfirmationMode
from .opcua_interface_param import OpcUaICommandType
from ....readwrite.typedef import *
from ....readwrite.write_interface_wrap import WriteInterfaceWrapper
from ....readwrite.read_interface_wrap import ReadInterfaceWrapper

from ....utils.numeric_utils import is_approximately_equal
from ....utils.logger import setup_logger

logger = setup_logger("driver.protocols.opcual_interface",use_async=True)
# 尝试导入opcua库，如果不存在则抛出运行时错误
try:
    from opcua import Client, ua
    from opcua.ua.uaerrors import UaError
    OPCUA_AVAILABLE = True
except ImportError:
    OPCUA_AVAILABLE = False
    logger.warning("Python OPC UA库未安装，OPC UA功能将不可用。请运行 'pip install opcua' 安装。")




class OpcUaInterface(CommunicationInterface):
    """OPC UA通信接口"""

    def __init__(self, config: Dict[str, Any]):
        """
        初始化OPC UA接口

        Args:
            config: 配置字典，应包含以下键值：
                - server_url: OPC UA服务器地址 (如 "opc.tcp://localhost:4840")
                - username: 用户名（可选）
                - password: 密码（可选）
                - security_policy: 安全策略（可选，默认为 "None"）
                - security_mode: 安全模式（可选，默认为 "None"）
                - certificate_path: 证书路径（可选）
                - private_key_path: 私钥路径（可选）
                - timeout: 超时时间（秒，默认为30)
                - auto_reconnect: 是否启用自动重连（可选，默认为True）
                - reconnect_attempts: 重连尝试次数（可选，默认为3）
                - reconnect_delay: 重连延迟时间（秒，默认为5）
        """
        if not OPCUA_AVAILABLE:
            raise RuntimeError("OPC UA库不可用。请安装 'opcua' 包以使用OPC UA功能。")

        super().__init__(config)
        self.protocol = CommunicationProtocol.OPC_UA
        self.server_url = config.get("server_url", "opc.tcp://localhost:4840")
        self.username = config.get("username")
        self.password = config.get("password")
        self.security_policy = config.get("security_policy", "None")
        self.security_mode = config.get("security_mode", "None")
        self.certificate_path = config.get("certificate_path")
        self.private_key_path = config.get("private_key_path")
        self.timeout = config.get("timeout", 30)
        self.auto_reconnect = config.get("auto_reconnect", True)
        self.reconnect_attempts = config.get("reconnect_attempts", 3)
        self.reconnect_delay = config.get("reconnect_delay", 5)
        self.client = None
        self._subscriptions = {}  # 存储活动的订阅
        self._reconnecting = False  # 正在重连标志
        self._session_state = defaultdict(dict)  # 存储会话状态信息，支持多个操作类型
        #线程安全处理
        self._lock = threading.RLock()  # 可重入锁，用于保护关键资源
        self._client_lock = threading.RLock()  # 用于保护客户端连接操作
        self._session_lock = threading.RLock()  # 用于保护会话状态
        self._subscription_lock = threading.RLock()  # 用于保护订阅操作
        self._read_lock = threading.RLock()  # 用于保护读操作
        self._write_lock = threading.RLock()  # 用于保护写操作
        self._read_batch_lock = threading.RLock()  # 用于保护批量读操作
        #服务命名空间
        self.namespace_index = 3  # 服务器命名空间索引
        self.address_prefix = "s"  # 地址变量前缀


    def _get_session_state(self, operation: str, session_id: Optional[str] = None) -> Dict:
        """
        获取特定操作和会话的状态

        Args:
            operation: 操作类型
            session_id: 会话ID（可选）

        Returns:
            Dict: 会话状态信息
        """
        key = f"{operation}_{session_id}" if session_id else operation
        with self._session_lock:
            return self._session_state[key]

    def _clear_session_state(self, operation: str, session_id: Optional[str] = None) -> None:
        """
        清除特定操作和会话的状态

        Args:
            operation: 操作类型
            session_id: 会话ID（可选）
        """
        key = f"{operation}_{session_id}" if session_id else operation
        with self._session_lock:
            if key in self._session_state:
                del self._session_state[key]

    def _save_session_state(self, operation: str, session_id: Optional[str] = None, **kwargs) -> None:
        """
        保存会话状态信息，用于重连后恢复

        Args:
            operation: 操作类型
            session_id: 会话ID（可选）
            **kwargs: 操作相关参数
        """
        key = f"{operation}_{session_id}" if session_id else operation
        with self._session_lock:
            self._session_state[key] = {
                'operation': operation,
                'params': kwargs
            }
        logger.debug(f"保存会话状态: {operation}" + (f"_{session_id}" if session_id else ""))

    def _save_operation_progress(self, operation: str, progress_data: Dict, session_id: Optional[str] = None) -> None:
        """
        保存操作进度信息，用于重连后恢复到断点

        Args:
            operation: 操作类型
            progress_data: 进度数据
            session_id: 会话ID（可选）
        """
        key = f"{operation}_{session_id}" if session_id else operation
        with self._session_lock:
            self._session_state[key]['operation'] = operation
            self._session_state[key]['progress'] = progress_data
        logger.debug(f"保存操作进度: {operation} - {progress_data}")

    def _ensure_connected(self) -> bool:
        """
        确保客户端已连接，如果未连接且启用了自动重连则尝试重连

        Raises:
            ConnectionError: 如果未连接且自动重连失败
        """
        with self._client_lock:
            self.is_connected: bool = self.is_available()  # 检查连接状态
            if not self.is_connected:
                if self.auto_reconnect and not self._reconnecting:
                    logger.warning("检测到连接断开，尝试自动重连...")
                    if not self._reconnect():
                        raise ConnectionError("OPC UA未连接且自动重连失败")
                else:
                    raise ConnectionError("OPC UA未连接")
        return True

    def _thread_safe_method(lock_name:str)-> Callable[[Callable[..., Any]], Callable[..., Any]]:
        """装饰器：为方法提供线程安全保护，使用指定的锁"""
        def decorator(func)-> Callable[..., Any]:
            @wraps(func)
            def wrapper(self, *args, **kwargs)-> Any:
                lock = getattr(self, lock_name)
                with lock:
                    return func(self, *args, **kwargs)
            return wrapper
        return decorator

    def connect(self) -> bool:
        """
        建立OPC UA连接

        Returns:
            bool: 连接是否成功
        """
        with self._client_lock:
            try:
                if self.is_connected:
                    logger.warning("OPC UA客户端已连接")
                    return True

                # 创建OPC UA客户端
                self.client = Client(self.server_url, timeout=self.timeout)

                # 配置安全策略
                if self.security_policy != "None":
                    security_string = f"Basic256Sha256,{self.security_mode}"
                    if self.certificate_path and self.private_key_path:
                        security_string += f",{self.certificate_path},{self.private_key_path}"
                    self.client.set_security_string(security_string)
                    logger.info(f"配置安全策略: {security_string}")
                else:
                    logger.info("使用无安全策略连接")

                # 设置认证信息
                if self.username and self.password:
                    self.client.set_user(self.username)
                    self.client.set_password(self.password)
                    logger.info(f"使用用户名 '{self.username}' 进行认证")
                else:
                    logger.info("使用匿名认证")

                # 连接到服务器
                self.client.connect()
                self.is_connected = True
                self._reconnecting = False

                logger.info(f"成功连接到OPC UA服务器: {self.server_url}")

                return True
            except UaError as e:
                logger.error(f"OPC UA连接错误: {e}")
                self.is_connected = False
                return False
            except Exception as e:
                logger.error(f"连接OPC UA服务器时发生未知错误: {e}")
                self.is_connected = False
                return False


    def disconnect(self) -> bool:
        """
        断开OPC UA连接

        Returns:
            bool: 断开连接是否成功
        """
        with self._client_lock:
            try:
                # 取消所有订阅
                with self._subscription_lock:
                    for sub_id, subscription in list(self._subscriptions.items()):
                        try:
                            subscription.delete()
                            del self._subscriptions[sub_id]
                        except Exception as e:
                            logger.warning(f"删除订阅 {sub_id} 失败: {e}")

                # 断开客户端连接
                if self.client:
                    self.client.disconnect()
                    self.client = None

                self.is_connected = False
                self._reconnecting = False
                with self._session_lock:
                    self._session_state.clear()  # 清除所有会话状态

                logger.info("成功断开OPC UA连接")
                return True
            except Exception as e:
                logger.error(f"断开OPC UA连接时发生错误: {e}")
                return False
    @_thread_safe_method('_client_lock')
    def _reconnect(self) -> bool:
        """
        重新连接到OPC UA服务器

        Returns:
            bool: 重连是否成功
        """

        if self._reconnecting:
            logger.debug("已在重连过程中，跳过重复重连请求")
            return False

        self._reconnecting = True
        logger.info("开始尝试重新连接到OPC UA服务器...")

        for attempt in range(self.reconnect_attempts):
            try:
                logger.debug(f"重连尝试 {attempt + 1}/{self.reconnect_attempts}")
                if self.connect():
                    logger.info("OPC UA服务器重连成功")
                    with self._client_lock:
                        self._reconnecting = False
                    return True
                else:
                    logger.warning(f"重连尝试 {attempt + 1} 失败")
            except Exception as e:
                logger.error(f"重连尝试 {attempt + 1} 发生异常: {e}")

            if attempt < self.reconnect_attempts - 1:
                logger.info(f"等待 {self.reconnect_delay} 秒后进行下一次重连尝试...")
                time.sleep(self.reconnect_delay)

        logger.error("所有重连尝试均已失败")
        self._reconnecting = False

        return False


    def set_namespaces(self, namespace_index: int,prefix: str = "s") -> bool:
        """
        设置命名空间索引，当访问时未包含ns命名空间时自动添加，
        Args:
            namespace_index: 命名空间索引 2,3,4,5...
            prefix: 地址前缀 "i"  "s"
        Returns:
                bool: 是否设置成功
        """
        with self._client_lock:
            try:
                self.namespace_index = namespace_index
                self.address_prefix=prefix
                return True
            except Exception as e:
                logger.error(f"设置命名空间列表失败: {e}")
                return False

    def get_namespaces(self) -> tuple[int, str]:
        """
        获取命名空间
        Returns:
                int: 命名空间索引
                str: 地址前缀
        """
        with self._client_lock:
            try:
                return (self.namespace_index,self.address_prefix)
            except Exception as e:
                logger.error(f"获取命名空间列表失败: {e}")
                return (-1,"")

    def is_available(self) -> bool:
        """
        检查OPC UA连接是否可用,如果连接不可用，尝试自动重连,用于心跳诊断
        需要更新连接状态 is_connected
        Returns:
            bool: 连接是否可用
        """
        with self._client_lock:
            if not self.is_connected or not self.client:
                if self.auto_reconnect and not self._reconnecting:
                    logger.warning("检测到连接断开，尝试自动重连...")
                    if self._reconnect():
                        logger.info("自动重连成功")
                    else:
                        logger.error("自动重连失败")
                        return False
                else:
                    return False
            if not self.client:
                raise ConnectionError("OPC UA未连接")
            try:
                # 尝试读取服务器状态节点以检查连接
                server_state: Node = self.client.get_node(ua.NodeId(ua.ObjectIds.Server_ServerStatus_State))
                server_state.get_value()
                return True
            except Exception as e:
                logger.debug(f"连接检查失败: {e}")
                # 尝试另一种方式检查连接可用性
                try:
                    # 尝试读取服务器对象节点
                    server_node = self.client.get_node(ua.NodeId(ua.ObjectIds.Server))
                    server_node.get_browse_name()
                    return True
                except Exception as e2:
                    logger.debug(f"备用连接检查也失败: {e2}")
                    self.is_connected = False
                    return False

    def _prepare_write_value(self, node_id: str, value: Any) -> bool:
        """
        准备并设置节点值

        Args:
            node_id: 节点ID
            value: 要写入的值

        Returns:
            bool: 写入是否成功
        """

        if not self.client:
            raise ConnectionError("OPC UA未连接")
        if "ns=" not in str(node_id):
            node_id = f"ns={self.namespace_index};{self.address_prefix}={node_id}"

        node = self.client.get_node(node_id)
        if isinstance(value, dict):
            # 如果值是一个字典，包含value和data_type
            data_value = ua.DataValue(ua.Variant(value["value"], getattr(ua.VariantType, value.get("data_type", "Int32"))))
            node.set_value(data_value)
            logger.info(f"节点{node_id}写入值: {value}类型: {value.get('data_type', 'Int32')}")
        else:
            # 直接设置值，使用DataValue和Variant避免BadWriteNotSupported错误
            data_type = self._get_node_dataType(node)
            if data_type=="Null":
                data_value = ua.DataValue(ua.Variant(value))
            else:
                data_value = ua.DataValue(ua.Variant(value, getattr(ua.VariantType, data_type)))

            node.set_value(data_value)
            logger.info(f"节点{node_id}写入值: {value}类型: {type(value)}")
        return True


    def _get_node_dataType(self,node)->str:
        """获取节点信息
        Args:
            node: 节点
        Returns:
            str: 节点数据类型 Null不可用
        Raises:
            Exception: 获取节点信息失败
        """

        try:
            # 获取属性
            attrs = node.get_attributes([
                ua.AttributeIds.Value,
                ua.AttributeIds.NodeClass
            ])

            # 提取属性值
            value_dv = attrs[0]  # DataValue
            node_class = attrs[1].Value.Value  # NodeClass

            # 判断是否是变量（有值）
            has_value = node_class in (ua.NodeClass.Variable, ua.NodeClass.VariableType)

            # 获取数据类型
            if has_value and value_dv.StatusCode.is_good():
                try:
                    # 尝试获取 VariantType
                    variant_type = value_dv.Value.VariantType
                    data_type = variant_type.name if variant_type != ua.VariantType.Null else "Null"
                except:
                    data_type = "Null"
            else:
                data_type = "Null"  # 如对象、方法等无值节点

            return data_type

        except Exception as e:
            logger.error(f"获取节点信息失败: {e}")
            return "Null"

    def _confirm_write_operation(self, node_id: str, expected_value: Any) -> bool:
        """
        验证写入操作是否成功

        Args:
            node_id: 节点ID
            expected_value: 期望值

        Returns:
            bool: 验证是否成功
        """
        try:
            if not self.client:
                raise ConnectionError("OPC UA未连接")

            if "ns=" not in str(node_id):
                node_id = f"ns={self.namespace_index};{self.address_prefix}={node_id}"

            node = self.client.get_node(node_id)
            verified_value = node.get_value()
            # 检查写入的值与读取的值是否一致
            if isinstance(expected_value, dict):
                expected_value = expected_value["value"]

            # 检查是否相等，使用工具函数进行近似比较
            if is_approximately_equal(verified_value, expected_value):
                logger.debug(f"写入验证成功: 节点 {node_id} 的值已确认为 {expected_value}")
                return True
            else:
                logger.warning(f"写入验证失败: 节点 {node_id} 期望值 {expected_value}，实际读取值 {verified_value}")
                return False
        except Exception as e:
            logger.error(f"写入验证过程中发生错误: {e}")
            return False

    def _retry_write_operation(self, node_id: str, value: Any, confirm: bool = False) -> bool:
        """
        重试写入操作

        Args:
            node_id: 节点ID
            value: 要写入的值
            confirm: 是否启用验证

        Returns:
            bool: 写入是否成功
        """
        expected_value = value["value"] if isinstance(value, dict) else value

        # 执行写入操作
        if not self._prepare_write_value(node_id, value):
            return False

        # 如果启用验证，则执行验证
        if confirm:
            if not self._confirm_write_operation(node_id, expected_value):
                return False

        return True



    @_thread_safe_method('_write_lock')
    def send_data(self, data: Any, **kwargs) -> Any:
        """
        发送数据（写入节点值）

        Args:
            data: 要写入的数据格式为字典，键为节点ID，值为要写入的值{"node_id": {"value": value, "data_type": type}}
            **kwargs: 其他参数
                - confirm: 是否启用写入后验证（默认为False）
                - retry: 重试次数（默认为3）
                - retry_delay: 重试间隔（秒，默认为1）
                - _reply: 是否保留之前已经执行的操作（默认为False）

        Returns:
            Any: 操作结果
        """
        # 保存操作状态用于重连后恢复
        if not kwargs.get('_reply', False):
            self._clear_session_state('send_data')
        self._save_session_state('send_data', data=data, **kwargs)

        self._ensure_connected()

        confirm = kwargs.get("confirm", False)
        retry_attempts = kwargs.get("retry", 3) if kwargs.get("retry") is not None else 3
        retry_delay = kwargs.get("retry_delay", 1)

        # 检查是否有保存的进度信息
        session_state = self._get_session_state('send_data')
        progress = session_state.get('progress', {})
        processed_nodes = set(progress.get('processed_nodes', []))

        for attempt in range(retry_attempts):
            try:
                results = {}
                failed_nodes = {}  # 记录失败的节点，用于重试

                if isinstance(data, dict):
                    # 处理未完成的节点
                    for node_id, value in data.items():
                        # 跳过已经处理过的节点
                        if node_id in processed_nodes:
                            continue

                        expected_value = value["value"] if isinstance(value, dict) else value

                        # 执行写入操作
                        with self._client_lock:
                            write_success = self._prepare_write_value(node_id, value)

                        if write_success:
                            # 如果启用验证，则执行一步写入一步验证策略
                            if confirm:
                                with self._client_lock:
                                    verification_success = self._confirm_write_operation(node_id, expected_value)
                                if verification_success:
                                    results[node_id] = True
                                else:
                                    failed_nodes[node_id] = value
                                    write_success = False
                            else:
                                results[node_id] = True
                        else:
                            failed_nodes[node_id] = value

                        # 保存当前操作进度，用于重连后恢复
                        if write_success:
                            processed_nodes.add(node_id)
                            self._save_operation_progress('send_data', {'processed_nodes': list(processed_nodes)})

                    # 如果有验证失败的节点且还有重试机会，则进行重试
                    if failed_nodes and attempt < retry_attempts - 1:
                        logger.info(f"发送数据部分失败，{retry_delay}秒后进行第{attempt + 2}次重试")
                        time.sleep(retry_delay)
                        # 更新data为失败的节点，以便重试
                        data = failed_nodes
                        continue
                    elif failed_nodes:
                        # 最后一次尝试仍有失败节点，抛出异常
                        raise Exception(f"发送数据失败，以下节点未能成功写入: {list(failed_nodes.keys())}")

                    # 操作成功完成，清除会话状态
                    self._clear_session_state('send_data')
                    return results
                else:
                    logger.error("数据格式错误，应为字典格式")
                    raise ValueError("数据格式错误，应为字典格式")
            except ValueError as e:
                # 捕获数据格式错误,不进行重试
                raise Exception(f"数据格式错误，应为字典格式: {e}")
            except Exception as e:
                logger.error(f"发送数据时发生错误: {e}")
                # 检查是否是连接异常，如果是且启用了自动重连则尝试重连
                if self.auto_reconnect and not self._reconnecting and not self.is_available():
                    logger.warning("发送数据时发生异常，尝试自动重连...")
                    if self._reconnect():
                        # 重连成功后重新尝试发送数据
                        if attempt < retry_attempts - 1:
                            logger.info(f"重连成功，{retry_delay}秒后进行第{attempt + 2}次重试")
                            time.sleep(retry_delay)
                            continue
                        # 重连成功但已达到最大重试次数，重新抛出异常
                        raise
                else:
                    if attempt < retry_attempts - 1:
                        logger.info(f"发送数据失败，{retry_delay}秒后进行第{attempt + 2}次重试")
                        time.sleep(retry_delay)
                        continue
                    # 已达到最大重试次数，重新抛出异常
                    raise

        raise Exception(f"发送数据失败，已重试{retry_attempts}次")

    @_thread_safe_method('_lock')
    def receive_data(self, timeout: Optional[float] = None) -> Any:
        """
        接收数据（此方法在OPC UA中不常用，因为通常使用订阅机制）

        Args:
            timeout: 超时时间（秒）

        Returns:
            Any: 接收到的数据
        """
        # 保存操作状态用于重连后恢复
        self._save_session_state('receive_data', timeout=timeout)

        logger.warning("receive_data方法在OPC UA中不常用，请考虑使用订阅机制")
        # 操作完成，清除会话状态
        self._clear_session_state('receive_data')
        # 这里简单返回None，实际应用中可能需要实现特定的读取逻辑
        return None

    def _execute_read_command(self, params: Dict) -> Any:
        """
        执行读取命令

        Args:
            params: 命令参数

        Returns:
            Any: 读取的数据
        """
        node_id:str = params.get("node_id", "")
        if not node_id:
            raise ValueError("读取操作需要提供node_id参数")
        if self.client is None:
            raise
        if "ns=" not in str(node_id):
            node_id = f"ns={self.namespace_index};{self.address_prefix}={node_id}"

        node = self.client.get_node(node_id)
        value = node.get_value()
        logger.info(f"节点{node_id}的值为: {value}")
        return value


    def _execute_write_command(self, params: Dict) -> bool:
        """
        执行写入命令

        Args:
            params: 命令参数
            - node_id: 节点ID
            - value: 要写入的值
            - confirm: 是否启用验证（默认为False）

        Returns:
            bool: 写入是否成功

        """
        node_id:str = params.get("node_id", "")
        value = params.get("value")
        confirm = params.get("confirm", False)
        # 执行写入操作
        if not self._prepare_write_value(node_id, value):
            return False

        # 如果启用验证，则执行验证
        if confirm:
            if not self._confirm_write_operation(node_id, value):
                return False

        return True


    def _execute_browse_command(self, params: Dict) -> List[Dict]:
        """
        执行浏览命令

        Args:
            params: 命令参数

        Returns:
            List[Dict]: 浏览结果
        """
        node_id = params.get("node_id", "i=85")  # 默认浏览根节点
        if self.client is None:
            raise
        if "ns=" not in str(node_id):
            node_id = f"ns={self.namespace_index};{self.address_prefix}={node_id}"

        node = self.client.get_node(node_id)
        children = node.get_children()

        result = []
        for child in children:
            result.append({
                "node_id": child.nodeid.to_string(),
                "browse_name": child.get_browse_name().to_string(),
                "display_name": child.get_display_name().Text,
                "node_class": child.get_node_class().name
            })
        return result

    def _execute_call_command(self, params: Dict) -> Any:
        """
        执行方法调用命令

        Args:
            params: 命令参数

        Returns:
            Any: 调用结果
        """
        method_id = params.get("method_id")
        parent_id = params.get("parent_id")
        args = params.get("args", [])

        if not method_id or not parent_id:
            raise ValueError("调用方法需要提供method_id和parent_id参数")

        if self.client is None:
            raise ConnectionError("OPC UA未连接")

        #添加命名空间
        if "ns=" not in str(method_id):
            method_id = f"ns={self.namespace_index};{self.address_prefix}={method_id}"
        if "ns=" not in str(parent_id):
            parent_id = f"ns={self.namespace_index};{self.address_prefix}={parent_id}"

        parent = self.client.get_node(parent_id)
        method = self.client.get_node(method_id)
        # 修复方法调用，需要将参数转换为正确的类型，TODO带完善
        ua_args = []
        for arg in args:
            if isinstance(arg, int):
                ua_args.append(ua.Variant(arg, ua.VariantType.Int32))
            elif isinstance(arg, str):
                ua_args.append(ua.Variant(arg, ua.VariantType.String))
            elif isinstance(arg, float):
                ua_args.append(ua.Variant(arg, ua.VariantType.Float))
            elif isinstance(arg, bool):
                ua_args.append(ua.Variant(arg, ua.VariantType.Boolean))
            else:
                ua_args.append(ua.Variant(arg))
        return parent.call_method(method, *ua_args)

    @_thread_safe_method('_write_lock')
    def execute_command(self, command: str, params: Optional[Dict] = None) -> Any:
        """
        执行OPC UA命令

        Args:
            command: 命令类型，支持 "read", "write", "browse", "call"
            params: 命令参数，只支持单个节点操作
            - read: {"node_id": "节点ID"}
            - write: {"node_id": "节点ID", "value": {"value": 123, "data_type": "Int32"}, "confirm": 是否启用验证（默认为False）}
            - browse: {"node_id": "节点ID"}（默认浏览根节点）
            - call: {"method_id": "方法ID", "parent_id": "父节点ID", "args": ["参数1", "参数2"]}

        Returns:
            Any: 命令执行结果
        """
        # 保存操作状态用于重连后恢复
        params = params or {}
        if not params.get('_reply', False):
            self._clear_session_state('execute_command')
        self._save_session_state('execute_command', command=command, params=params)

        self._ensure_connected()

        params = params or {}
        retry_attempts = params.get("retry", 3) if params.get("retry") is not None else 3
        retry_delay = params.get("retry_delay", 1)
        for attempt in range(retry_attempts):
            try:
                command =command.value if isinstance(command, OpcUaICommandType) else command
                if command == "read":
                    with self._client_lock:
                        result = self._execute_read_command(params)
                    # 操作成功完成，清除会话状态
                    self._clear_session_state('execute_command')
                    return result

                elif command == "write":
                    with self._client_lock:
                        result = self._execute_write_command(params)
                    # 操作成功完成，清除会话状态
                    self._clear_session_state('execute_command')
                    return result

                elif command == "browse":
                    with self._client_lock:
                        result = self._execute_browse_command(params)
                    # 操作成功完成，清除会话状态
                    self._clear_session_state('execute_command')
                    return result

                elif command == "call":
                    with self._client_lock:
                        result = self._execute_call_command(params)
                    # 操作成功完成，清除会话状态
                    self._clear_session_state('execute_command')
                    return result

                else:
                    raise ValueError(f"不支持的命令: {command}")
            except ValueError as e:
                logger.error(f"发生异常，参数: {e}")
                raise Exception(f"发生异常，参数: {e}")# 不进行重试
            except Exception as e:
                logger.error(f"execute_command发送数据时发生错误: {e}")
                # 检查是否是连接异常，如果是且启用了自动重连则尝试重连
                if self.auto_reconnect and not self._reconnecting and not self.is_available():
                    logger.warning("发送数据时发生异常，尝试自动重连...")
                    if self._reconnect():
                        # 重连成功后重新尝试发送数据
                        if attempt < retry_attempts - 1:
                            logger.info(f"重连成功，{retry_delay}秒后进行第{attempt + 2}次重试")
                            time.sleep(retry_delay)
                            continue
                        # 重连成功但已达到最大重试次数，重新抛出异常
                        raise
                else:
                    if attempt < retry_attempts - 1:
                        logger.info(f"发送数据失败，{retry_delay}秒后进行第{attempt + 2}次重试")
                        time.sleep(retry_delay)
                        continue
                    # 已达到最大重试次数，重新抛出异常
                    raise
        # 如果所有重试都失败，抛出异常
        raise Exception(f"execute_command发送数据失败，已重试{retry_attempts}次")
    @_thread_safe_method('_read_batch_lock')
    def read_batch(self, addresses: List[Union[str, int]], **kwargs) -> Dict[Union[str, int], Any]:
        """
        批量读取节点值

        Args:
            addresses: 节点ID列表["ns=2;i=2", "ns=2;i=3", "ns=2;i=4", "ns=2;i=5"]
            **kwargs: 其他参数
                - retry: 重试次数（默认为3）
                - retry_delay: 重试间隔（秒，默认为1）
                - _reply: 是否保留之前已经执行的操作（默认为False）

        Returns:
            Dict[Union[str, int], Any]: 节点ID和对应值的映射
        """
        logger.debug(f"准备批量读取节点: addresses={addresses}, kwargs={kwargs}")

        # 保存操作状态用于重连后恢复
        if not kwargs.get('_reply', False):
            self._clear_session_state('read_batch')
        self._save_session_state('read_batch', addresses=addresses, **kwargs)

        self._ensure_connected()

        retry_attempts = kwargs.get("retry", 3) if kwargs.get("retry") is not None else 3
        retry_delay = kwargs.get("retry_delay", 1)
        if not self.client:
            raise ConnectionError("OPC UA未连接")
        for attempt in range(retry_attempts):
            try:
                with self._client_lock:
                    # 将节点ID转换为节点对象
                    nodes = []
                    for node_id in addresses:
                        if "ns=" not in str(node_id):
                            node_id = f"ns={self.namespace_index};{self.address_prefix}={node_id}"
                        nodes.append(self.client.get_node(node_id))
                    values = self.client.get_values(nodes)

                    # 构造返回结果
                    results = {}
                    for i, node_id in enumerate(addresses):
                        results[node_id] = values[i]

                    # 操作成功完成，清除会话状态
                    self._clear_session_state('read_batch')

                    logger.info(f"成功批量读取节点: count={len(results)}, results={results}")
                    return results

            except Exception as e:
                logger.error(f"批量读取时发生异常: {e}")
                # 检查是否是连接异常，如果是且启用了自动重连则尝试重连
                if self.auto_reconnect and not self._reconnecting and not self.is_available():
                    logger.warning("批量读取时发生异常，尝试自动重连...")
                    if not self._reconnect():
                        logger.error("自动重连失败")
                        raise

                    # 重新尝试批量读取
                    if attempt < retry_attempts - 1:
                        logger.info(f"重连成功，{retry_delay}秒后进行第{attempt + 2}次重试")
                        time.sleep(retry_delay)
                        continue
                    else:
                        logger.error("批量读取失败，已重试最大次数")
                        # 重连成功但已达到最大重试次数，重新抛出异常
                        raise
                else:
                    if attempt < retry_attempts - 1:
                        logger.info(f"批量读取失败，{retry_delay}秒后进行第{attempt + 2}次重试")
                        time.sleep(retry_delay)
                        continue
                    else:
                        logger.error("批量读取失败，已重试最大次数")
                        # 已达到最大重试次数，重新抛出异常
                        raise

        raise Exception(f"批量读取失败，已重试{retry_attempts}次")

    def _confirm_write_batch(self, data: Dict[Union[str, int], Any], **kwargs) -> Tuple[bool, Dict]:
        """
        验证批量写入操作

        Args:
            data: 要验证的数据，应为字典格式 {"ns=2;i=2": 111}
            **kwargs: 其他参数
                - retry: 重试次数（默认为3）
                - retry_delay: 重试间隔（秒，默认为1）

        Returns:
            Tuple[bool, Dict]: (验证是否成功, 失败的数据)
        """

        retry_attempts = kwargs.get("retry", 3) if kwargs.get("retry") is not None else 3
        retry_delay = kwargs.get("retry_delay", 1)
        if not self.client:
            raise ConnectionError("OPC UA未连接")
        for attempt in range(retry_attempts):
            try:
                # 批量读取验证写入的值
                nodes = []
                original_values = []
                node_ids = []

                for node_id, value in data.items():
                    node_id_new=node_id
                    if "ns=" not in str(node_id):
                        node_id_new = f"ns={self.namespace_index};{self.address_prefix}={node_id}"
                    nodes.append(self.client.get_node(node_id_new))
                    if isinstance(value, dict):
                        original_values.append(value["value"])
                    else:
                        original_values.append(value)

                    node_ids.append(node_id)

                # 批量读取验证写入的值
                with self._client_lock:
                    verified_values = self.client.get_values(nodes)

                # 检查写入的值与读取的值是否一致
                verification_passed = True
                failed_data = {}

                for i, node_id in enumerate(node_ids):
                    # 使用工具函数进行近似相等比较
                    if not is_approximately_equal(verified_values[i], original_values[i]):
                        logger.warning(f"批量写入验证失败: 节点 {node_id} 期望值 {original_values[i]}，实际读取值 {verified_values[i]}")
                        verification_passed = False
                        failed_data[node_id] = original_values[i]

                if verification_passed:
                    logger.debug(f"批量写入验证成功: 所有 {len(data)} 个节点的值已确认")

                return verification_passed, failed_data

            except Exception as e:
                logger.error(f"批量写入验证时发生异常: {e}")
                # 检查是否是连接异常，如果是且启用了自动重连则尝试重连
                if self.auto_reconnect and not self._reconnecting and not self.is_available():
                    logger.warning("批量写入验证时发生异常，尝试自动重连...")
                    if self._reconnect():
                        # 重连成功后重新尝试批量写入
                        if attempt < retry_attempts - 1:
                            logger.info(f"重连成功，{retry_delay}秒后进行第{attempt + 2}次重试")
                            time.sleep(retry_delay)
                            continue
                        raise Exception(f"批量写入验证失败，已重试{retry_attempts}次")
                else:
                    if attempt < retry_attempts - 1:
                        logger.info(f"批量写入验证失败，{retry_delay}秒后进行第{attempt + 2}次重试")
                        time.sleep(retry_delay)
                        continue
                    raise Exception(f"批量写入验证失败，已重试{retry_attempts}次")
        raise Exception(f"批量写入验证失败，已重试{retry_attempts}次")


    @_thread_safe_method('_write_lock')
    def write_batch(self, data: Dict[Union[str, int], Any], **kwargs) -> bool:
        """
        批量写入节点值

        Args:
            data: 节点ID和值的映射{"ns=2;i=2": {"value": 111, "data_type": "Int32"}}
            **kwargs: 其他参数
                - confirm: 是否启用写入后验证（默认为False）
                - retry: 重试次数（默认为3）
                - retry_delay: 重试间隔（秒，默认为1）
                - _reply: 是否保留之前已经执行的操作（默认为False）

        Returns:
            bool: 写入是否成功
        """
        logger.debug(f"准备批量写入节点: data={data}, kwargs={kwargs}")

        # 保存操作状态用于重连后恢复
        if not kwargs.get('_reply', False):
            self._clear_session_state('write_batch')
        self._save_session_state('write_batch', data=data, **kwargs)

        self._ensure_connected()

        confirm = kwargs.get("confirm", False)
        retry_attempts = kwargs.get("retry", 3) if kwargs.get("retry") is not None else 3
        retry_delay = kwargs.get("retry_delay", 1)

        # 检查是否有保存的进度信息
        session_state = self._get_session_state('write_batch')
        progress = session_state.get('progress', {})
        processed_nodes = set(progress.get('processed_nodes', []))
        failed_nodes = progress.get('failed_nodes', {})

        # 如果有失败节点信息，从失败节点继续
        if failed_nodes:
            logger.info(f"从 {len(failed_nodes)} 个失败节点继续批量写入操作")
            data = failed_nodes

        if not self.client:
            raise ConnectionError("OPC UA未连接")

        for attempt in range(retry_attempts):
            try:
                node_objects = []
                values = []
                items = list(data.items())
                current_failed_nodes = {}

                for i, (node_id, value) in enumerate(items):
                    # 跳过已经处理过的成功节点
                    if node_id in processed_nodes:
                        continue

                    # 将节点ID转换为节点对象
                    if "ns=" not in str(node_id):
                        node_id = f"ns={self.namespace_index};{self.address_prefix}={node_id}"

                    node = self.client.get_node(node_id)
                    node_objects.append(node)
                    if isinstance(value, dict):
                        # 如果值是一个字典，包含value和data_type
                        data_value = ua.DataValue(ua.Variant(value["value"], getattr(ua.VariantType, value.get("data_type", "Int32"))))
                        values.append(data_value)
                    else:
                        # 直接设置值，使用DataValue和Variant避免BadWriteNotSupported错误
                        data_type = self._get_node_dataType(node)
                        if data_type =="Null":
                            data_value = ua.DataValue(ua.Variant(value))
                        else:
                            data_value = ua.DataValue(ua.Variant(value, getattr(ua.VariantType, data_type)))
                        values.append(data_value)

                # 执行批量写入操作
                if node_objects:  # 只有当有节点需要写入时才执行
                    with self._client_lock:
                        self.client.set_values(node_objects, values)
                    logger.debug(f"已完成批量写入操作: count={len(node_objects)}")

                # 验证写入结果
                verification_passed = True
                current_failed_nodes_confirm = {}

                # 只有当有节点需要验证时才执行
                if confirm and node_objects:
                    try:
                        with self._client_lock:
                            verification_passed, current_failed_nodes_confirm = self._confirm_write_batch(data, **kwargs)
                    except Exception as e:
                        logger.error(f"批量写入验证时发生异常: {e}")
                        return False


                # 合并失败节点
                all_failed_nodes = {**current_failed_nodes, **current_failed_nodes_confirm}

                # 更新处理进度
                for node_id, _ in items:
                    if node_id not in all_failed_nodes:
                        processed_nodes.add(node_id)

                # 保存当前操作进度，用于重连后恢复
                self._save_operation_progress('write_batch', {
                    'processed_nodes': list(processed_nodes),
                    'failed_nodes': all_failed_nodes
                })


                if all_failed_nodes:
                    data = all_failed_nodes  # 更新data为失败的节点
                    logger.error(f"批量写入失败: {len(all_failed_nodes)} 个节点写入失败")
                    raise Exception(f"批量写入失败: {len(all_failed_nodes)} 个节点写入失败")

                # 操作成功完成，清除会话状态
                self._clear_session_state('write_batch')

                logger.info(f"成功批量写入节点: count={len(items) - len(all_failed_nodes)}, failed={len(all_failed_nodes)}")
                return True

            except Exception as e:
                logger.error(f"批量写入时发生异常: {e}")
                # 检查是否是连接异常，如果是且启用了自动重连则尝试重连
                if self.auto_reconnect and not self._reconnecting and not self.is_available():
                    logger.warning("批量写入时发生异常，尝试自动重连...")
                    if self._reconnect():
                        # 重连成功后重新尝试批量写入
                        if attempt < retry_attempts - 1:
                            logger.info(f"重连成功，{retry_delay}秒后进行第{attempt + 2}次重试")
                            time.sleep(retry_delay)
                            continue
                        raise Exception(f"批量写入失败，已重试{retry_attempts}次")
                else:
                    if attempt < retry_attempts - 1:
                        logger.info(f"批量写入失败，{retry_delay}秒后进行第{attempt + 2}次重试")
                        time.sleep(retry_delay)
                        continue
                    raise Exception(f"批量写入失败，已重试{retry_attempts}次")
        raise Exception(f"批量写入失败，已重试{retry_attempts}次")

    @_thread_safe_method('_read_lock')
    def read_register(self, address: Union[str, int], data_type: str = "int", **kwargs) -> Any:
        """
        读取寄存器数据（节点值）

        Args:
            address: 节点ID
            data_type: 数据类型（未使用，OPC UA自动处理）
            **kwargs: 其他参数
                - retry: 重试次数（默认为3）
                - retry_delay: 重试间隔（秒，默认为1）

        Returns:
            Any: 读取的数据
        """
        try:
            logger.debug(f"准备读取寄存器: address={address}, data_type={data_type}, kwargs={kwargs}")

            # 构造传递给execute_command的参数
            params = {"node_id": address}
            params.update(kwargs)  # 将kwargs中的参数传递给execute_command

            result = self.execute_command("read", params)

            logger.info(f"成功读取寄存器: address={address}, value={result}")
            return result
        except Exception as e:
            logger.error(f"读取寄存器 {address} 失败: {e}")
            raise Exception(f"读取寄存器失败: {e}")

    @_thread_safe_method('_write_lock')
    def write_register(self, address: Union[str, int], value: Any, data_type: str=None , **kwargs) -> bool:
        """
        写入寄存器数据（节点值）

        Args:
            address: 节点ID
            value: 要写入的值
            data_type: 数据类型（如"Int32", "Boolean", "Double"等）
            data_type: 数据类型（未使用，OPC UA自动处理,通过查询服务器处理）
            **kwargs: 其他参数
                - confirm: 是否启用写入后验证（默认为False）
                - retry: 重试次数（默认为3）
                - retry_delay: 重试间隔（秒，默认为1）

        Returns:
            bool: 写入是否成功
        """
        try:
            logger.debug(f"准备写入寄存器: address={address}, value={value}, data_type={data_type}, kwargs={kwargs}")

            # 构造传递给execute_command的参数
            if data_type is not None:
                # 如果指定了数据类型，则构造字典形式的值
                params = {"node_id": address, "value": {"value": value, "data_type": data_type}}
            else:
                # 否则直接传递值
                params = {"node_id": address, "value": value}
            params.update(kwargs)  # 将kwargs中的参数传递给execute_command

            result = self.execute_command("write", params)

            logger.info(f"成功写入寄存器: address={address}, value={value}, data_type={data_type}, result={result}")
            return result
        except Exception as e:
            logger.error(f"写入寄存器 {address} 失败: {e}")
            raise Exception(f"写入寄存器失败: {e}")

    @_thread_safe_method('_subscription_lock')
    def subscribe(self, address: Union[str, int], callback: Callable[[Any], None], **kwargs) -> str:
        """
        订阅数据变化

        Args:
            address: 要订阅的节点ID
            callback: 回调函数
            **kwargs: 其他参数，如采样间隔(sampling_interval)
                - _reply: 是否保留之前已经执行的操作（默认为False）

        Returns:
            str: 订阅ID
        """
        # 保存操作状态用于重连后恢复
        if not kwargs.get('_reply', False):
            self._clear_session_state('subscribe')
        self._save_session_state('subscribe', address=address, callback=callback, **kwargs)

        self._ensure_connected()

        if "ns=" not in str(address):
            address = f"ns={self.namespace_index};{self.address_prefix}={address}"
            logger.debug(f"订阅自动补全节点地址: {address}")

        if not self.client:
            raise ConnectionError("OPC UA未连接")

        try:
            sampling_interval = kwargs.get("sampling_interval", 1000)  # 默认1000ms

            # 创建订阅处理类
            class SubscriptionHandler:
                def __init__(self, callback_func):
                    self.callback = callback_func

                def datachange_notification(self, node, val, data):
                    try:
                        self.callback(val)
                    except Exception as e:
                        logger.error(f"订阅回调执行失败: {e}")

                def event_notification(self, event):
                    pass  # 我们不处理事件通知

            # 创建订阅
            handler = SubscriptionHandler(callback)
            with self._client_lock:
                subscription = self.client.create_subscription(sampling_interval, handler)
                node = self.client.get_node(address)
                handle = subscription.subscribe_data_change(node)

            # 存储订阅信息
            subscription_id = str(handle)
            with self._subscription_lock:
                self._subscriptions[subscription_id] = subscription

            logger.debug(f"成功订阅节点 {address}，订阅ID: {subscription_id}")
            return subscription_id
        except Exception as e:
            # 检查是否是连接异常，如果是且启用了自动重连则尝试重连
            if self.auto_reconnect and not self._reconnecting and not self.is_available():
                logger.warning("订阅时发生异常，尝试自动重连...")
                if self._reconnect():
                    # 重连成功后重新尝试订阅
                    return self.subscribe(address, callback, **kwargs)
            logger.error(f"订阅节点 {address} 失败: {e}")
            raise

    @_thread_safe_method('_subscription_lock')
    def unsubscribe(self, subscription_id: str) -> bool:
        """
        取消订阅

        Args:
            subscription_id: 订阅ID

        Returns:
            bool: 取消订阅是否成功
        """
        with self._subscription_lock:
            try:
                if subscription_id in self._subscriptions:
                    subscription = self._subscriptions[subscription_id]
                    with self._client_lock:
                        subscription.delete()
                    del self._subscriptions[subscription_id]
                    logger.debug(f"成功取消订阅: {subscription_id}")
                    return True
                else:
                    logger.warning(f"未找到订阅: {subscription_id}")
                    return False
            except Exception as e:
                logger.error(f"取消订阅 {subscription_id} 失败: {e}")
                return False

    @_thread_safe_method('_lock')
    async def async_send_data(self, data: Any, **kwargs) -> Any:
        """
        异步发送数据

        Args:
            data: 要发送的数据
            **kwargs: 其他参数

        Returns:
            Any: 发送操作的结果
        """
        # 使用asyncio的默认事件循环运行同步方法
        loop = asyncio.get_event_loop()
        return await loop.run_in_executor(None, self.send_data, data, kwargs)

    @_thread_safe_method('_lock')
    async def async_receive_data(self, timeout: Optional[float] = None) -> Any:
        """
        异步接收数据

        Args:
            timeout: 超时时间

        Returns:
            Any: 接收到的数据
        """
        # 使用asyncio的默认事件循环运行同步方法
        loop = asyncio.get_event_loop()
        return await loop.run_in_executor(None, self.receive_data, timeout)

    @_thread_safe_method('_lock')
    def diagnose(self) -> Dict[str, Any]:
        """
        诊断设备状态

        Returns:
            Dict[str, Any]: 诊断信息
        """
        try:
            with self._client_lock:
                if not self.is_connected or not self.client:
                    return {"status": "disconnected", "error": "客户端未连接"}

                # 获取服务器状态
                try:
                    server_state_node = self.client.get_node(ua.NodeId(ua.ObjectIds.Server_ServerStatus_State))
                    server_state = server_state_node.get_value()
                    state_names = {
                        0: "Running",
                        1: "Failed",
                        2: "NoConfiguration",
                        3: "Suspended",
                        4: "Shutdown",
                        5: "Test",
                        6: "CommunicationFault",
                        7: "Unknown"
                    }
                    server_state_str = state_names.get(server_state, f"Unknown ({server_state})")
                except Exception:
                    server_state_str = "Unknown"

                # 获取服务器启动时间
                try:
                    start_time_node = self.client.get_node(ua.NodeId(ua.ObjectIds.Server_ServerStatus_StartTime))
                    start_time = start_time_node.get_value()
                except Exception:
                    start_time = "Unknown"

            with self._subscription_lock:
                active_subscriptions = len(self._subscriptions)

            return {
                "status": "connected",
                "server_url": self.server_url,
                "server_state": server_state_str,
                "server_start_time": start_time,
                "active_subscriptions": active_subscriptions,
                "security_policy": self.security_policy,
                "security_mode": self.security_mode,
                "auto_reconnect": self.auto_reconnect,
                "reconnect_attempts": self.reconnect_attempts,
                "reconnect_delay": self.reconnect_delay
            }
        except Exception as e:
            return {"status": "error", "error": str(e)}

    @_thread_safe_method('_read_lock')
    def read(self,
             address: FullReadType,
             *args: Union[
                 int,              # timeout (seconds)
                 float,            # timeout (seconds)
                 str,              # data_type or read_option
                 DataType,         # data type enum
                 ReadOption,       # read option enum
                 PreReaderType,    # pre-read processor function
                 PostReaderType,   # post-read processor function
                 Dict[str, Any]    # additional configuration
             ],
             timeout: Optional[Union[int, float]] = None,
             data_type: Optional[Union[str, DataType]] = None,
             sync: Optional[bool] = None,
             retry: Optional[int] = None,
             raise_on_error: Optional[bool] = None,
             quality_check: Optional[bool] = None,
             prereaders: Optional[List[PreReaderType]] = None,
             postreaders: Optional[List[PostReaderType]] = None,
             batch_size: Optional[int] = None,
             **kwargs: Any) -> Union[
                 Any,                     # single read result
                 List[Any],               # batch read results as list
                 Dict[AddressType, Any],  # batch read results as dict
                 Dict[str, Any]           # config read results
             ]:
        """
        读取数据,与设备接口一致用法，此处仅做说明，不补充注释

        Examples:
            # 单个读取：最简单的读取形式
            value = client.read("address1")

            # 带数据类型的单个读取：显式指定数据类型
            value = client.read(("address1", "int"))

            # 使用位置参数指定数据类型
            value = client.read("address1", "int")

            # 批量读取：通过列表一次读取多个地址
            values = client.read(["address1", "address2"])

            # 带配置的批量读取：通过元组列表指定多个地址和数据类型
            values = client.read([("address1", "int"), ("address2", "float")])

            # 带数据类型和选项的批量读取
            values = client.read([("address1", "int", "option1"), ("address2", "float", "option2")])

            # 使用关键字参数配置：通过关键字参数指定额外配置
            values = client.read("address1", timeout=10, data_type="int", sync=True)

            # 配置读取：通过配置字典执行读取操作
            result = client.read({
                "addresses": ["address1", "address2"],
                "data_type": "int"
            })

            # 带有所有可选参数的完整读取
            result = client.read(
                "address1",
                timeout=30,
                data_type=DataType.INT32,
                sync=True,
                retry=3,
                raise_on_error=True,
                quality_check=True
            )

        Note:
            1. 参数优先级：kwargs > args > defaults
            2. 支持的数据类型包括所有DataType枚举值
            3. 读取选项包括SYNC, ASYNC
            4. 当batch_size参数设置且读取数据量超过该值时，将自动分块处理
            5. prereaders在数据读取前执行，postreaders在数据读取后执行

        """
        try:
            return ReadInterfaceWrapper.read(self, address, *args, timeout=timeout, data_type=data_type, sync=sync, retry=retry, raise_on_error=raise_on_error,
                                            quality_check=quality_check, prereaders=prereaders, postreaders=postreaders, batch_size=batch_size, **kwargs)
        except Exception as e:
            logger.error(f"读取数据 {address} 失败: {e}")
            raise ValueError(f"读取数据 {address} 失败: {e}")

    @_thread_safe_method('_write_lock')
    def write(self,
              address: FullWriteType,
              value: Optional[Any] = None,
              *args: Union[
                  int,               # timeout (seconds)
                  float,             # timeout (seconds)
                  str,               # data_type or write_option
                  DataType,          # data type enum
                  WriteOption,       # write option enum
                  PreProcessorType,  # pre-write processor function
                  PostProcessorType, # post-write processor function
                  Dict[str, Any]     # additional configuration
              ],
              timeout: Optional[Union[int, float]] = None,
              data_type: Optional[Union[str, DataType]] = None,
              sync: Optional[bool] = None,
              confirm: Optional[bool] = None,
              retry: Optional[int] = None,
              raise_on_error: Optional[bool] = None,
              quality_check: Optional[bool] = None,
              preprocessors: Optional[List[PreProcessorType]] = None,
              postprocessors: Optional[List[PostProcessorType]] = None,
              batch_size: Optional[int] = None,
              transaction: Optional[bool] = None,
              **kwargs: Any) -> Union[
                  bool,                    # single write result
                  Dict[AddressType, bool], # batch write results
                  Dict[str, Any]           # config write results
              ]:
        """
        写入数据,与设备接口一致用法，此处仅做说明，不补充注释

        Examples:
            # 单个写入：最简单的写入形式
            client.write("address1", 100)

            # 带数据类型的单个写入：显式指定数据类型
            client.write(("address1", 100, "int"))

            # 使用位置参数指定数据类型
            client.write("address1", 100, "int")

            # 带数据类型和长度的单个写入：指定数据类型和额外参数（长度）
            client.write(("address1", 100, "int", 4))

            # 批量写入：通过字典一次写入多个地址
            client.write({"address1": 100, "address2": 200})

            # 带配置的批量写入：通过元组列表指定多个地址和数据类型
            client.write([("address1", 100, "int"), ("address2", 200, "float")])

            # 带数据类型和选项的批量写入
            client.write([("address1", 100, "int", "option1"), ("address2", 200, "float", "option2")])

            # 使用关键字参数配置：通过关键字参数指定额外配置
            client.write("address1", 100, timeout=10, data_type="int", sync=True)

            # 配置写入：通过配置字典执行写入操作
            client.write({
                "addresses": ["address1", "address2"],
                "values": [100, 200],
                "data_type": "int"
            })

            # 带有所有可选参数的完整写入
            client.write(
                "address1",
                100,
                timeout=30,
                data_type=DataType.INT32,
                sync=True,
                confirm=True,
                retry=3,
                raise_on_error=False,
                quality_check=True
            )

        Note:
            1. 参数优先级：kwargs > args > defaults
            2. 支持的数据类型包括所有DataType枚举值
            3. 写入选项包括SYNC, ASYNC
            4. 当batch_size参数设置且写入数据量超过该值时，将自动分块处理
            5. preprocessors在数据写入前执行，postprocessors在数据写入后执行

        """
        try:
           return WriteInterfaceWrapper.write(self, address, value, *args,timeout=timeout, data_type=data_type, sync=sync, confirm=confirm, retry=retry,
                                              raise_on_error=raise_on_error, quality_check=quality_check, preprocessors=preprocessors,
                                              postprocessors=postprocessors, batch_size=batch_size, transaction=transaction, **kwargs)
        except Exception as e:
            logger.error(f"写入数据 {address} 失败: {e}")
            raise ValueError(f"写入数据 {address} 失败: {e}")